package cn.doitedu.rtmk.demo6;

import cn.doitedu.rtmk.common.EventBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class EngineProcessFunction extends KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String> {


    final HashMap<String,RuleCalculator> calculatorHashMap = new HashMap<>();

    @Override
    public void processElement(EventBean eventBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {

        for (Map.Entry<String, RuleCalculator> entry : calculatorHashMap.entrySet()) {
            RuleCalculator ruleCalculator = entry.getValue();
            ruleCalculator.calc(eventBean,collector);

        }

    }

    @Override
    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.Context context, Collector<String> collector) throws Exception {

        // 收到的 ruleMetaBean ，有可能代表 管理平台 增加、修改、删除了规则参数

        // 如果操作是新增或更新后，则我们的逻辑相同，都是构造运算机，并放入运算机池
        if("+I".equals(ruleMetaBean.getOp()) || "+U".equals(ruleMetaBean.getOp())  ){

            // 构造运行机，并使用传入的规则参数来初始化该运算机对象
            RuleCalculator ruleCalculator = new RuleModel_1_Calculator();
            ruleCalculator.init(getRuntimeContext(),ruleMetaBean.getRuleParamJson());

            // 将初始化好的运算机，放入运算机池
            calculatorHashMap.put(ruleMetaBean.getRuleId(),ruleCalculator);

            log.warn("收到新增规则的请求，成功新增了规则运算机: {}",ruleMetaBean.getRuleId());

        }
        // 如果操作上 删除，则我们需要把该规则从运算机池移除
        else if("-D".equals(ruleMetaBean.getOp())){
            calculatorHashMap.remove(ruleMetaBean.getRuleId());

            log.warn("收到删除规则的请求，成功删除了规则运算机: {}",ruleMetaBean.getRuleId());

        }


    }



}
